package com.eric.log.mapred;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Hashtable;
import java.util.Locale;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.eric.log.mapred.LogBasicMapRed.BasicMapper;

public class PageViewMapRed {
	static Hashtable<String,String> ht = new Hashtable<>();
	
	public static class PageViewReducer extends Reducer<Text,Text,Text,Text> {
		public void reduce (Text key,Iterable<Text> value,Context context) throws IOException, InterruptedException {
			int counter = 0;
			Date timeStamp = null,prevTimeStamp = null;
			long timeOnPage = 0;
			
			for (Text v : value) {
				counter++;
				
				String[] entry = v.toString().replaceAll("\"", "").split(Config.FS);
				try {
					timeStamp = new SimpleDateFormat(Config.DateFormat,Locale.ENGLISH).parse(entry[Config.FN_DATE]+" "+entry[Config.FN_TIME]);
				} catch (ParseException e) {
					e.printStackTrace();
				}
				
				if (counter == 1) prevTimeStamp = timeStamp;
				if (counter > 1) timeOnPage = Math.abs(timeStamp.getTime() - prevTimeStamp.getTime());
				
				prevTimeStamp = timeStamp;
				
				context.write(key, new Text(v.toString()+Config.FS+timeOnPage));
			}
		}
	}
	
	public static class PageViewReducer4Mahout extends Reducer<Text,Text,Text,Text> {
		public void reduce (Text key,Iterable<Text> value,Context context) throws IOException, InterruptedException {
			int counter = 0;
			Date timeStamp = null,prevTimeStamp = null;
			long timeOnPage = 0;
			
			for (Text v : value) {
				counter++;
				
				String[] entry = v.toString().replaceAll("\"", "").split(Config.FS);
				try {
					timeStamp = new SimpleDateFormat(Config.DateFormat,Locale.ENGLISH).parse(entry[Config.FN_DATE]+" "+entry[Config.FN_TIME]);
				} catch (ParseException e) {
					e.printStackTrace();
				}
				
				if (counter == 1) prevTimeStamp = timeStamp;
				if (counter > 1) timeOnPage = Math.abs(timeStamp.getTime() - prevTimeStamp.getTime());
				
				prevTimeStamp = timeStamp;
				
				if (entry[Config.FN_URL].matches("/goods/[0-9]+\\.html") 
						&& !key.toString().contains("ptid.stid")
						&& key.toString().matches("[0-9a-z]+\\.[0-9a-z]+")) {
					String id = String.valueOf(key.toString().replaceAll("\"", "").split(".")[1].hashCode());
					String goods_id = entry[Config.FN_URL].replaceAll("\"", "").replaceAll("(/goods/)|(.html.*)|(\\?)", "");
					
					context.write(new Text(id), 
									new Text(goods_id+Config.FS+timeOnPage / 6000.00));
				}
			}
		}
	}
	
	public static void run (String input,String output) throws IOException, ClassNotFoundException, InterruptedException {
		LogBasicMapRed.setKeyFieldNum(Config.FN_UTM);
		
		Configuration conf = new Configuration();
		conf.set("mapred.textoutputformat.separator", ",");
	    Job job = new Job(conf,"PageView Map Reduce");
	    
	    job.setMapperClass(BasicMapper.class);
	    job.setReducerClass(PageViewReducer4Mahout.class);
	    
	    job.setJarByClass(VisitMapRed.class);
	    
	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(Text.class);
	    
	    FileInputFormat.addInputPath(job, new Path(input));
	    FileOutputFormat.setOutputPath(job, new Path(output));
	    
	    job.waitForCompletion(true);
	}
}
